package Log

import (
	"context"
	"encoding/json"
	"log"
	"sync"
	models "web_back_go/DAO/model"

	"github.com/go-redis/redis/v8"
	"xorm.io/xorm"
)

type Logs struct {
	Id        int    `xorm:"not null pk autoincr INT"`
	LogLevel  string `xorm:"VARCHAR(50)" json:"level"`
	Timestamp int64  `xorm:"BIGINT" json:"ts"`
	Caller    string `xorm:"VARCHAR(255)"`
	Msg       string `xorm:"VARCHAR(255)"`
	Method    string `xorm:"VARCHAR(10)"`
	Path      string `xorm:"VARCHAR(255)"`
	Query     string `xorm:"VARCHAR(255)"`
	Body      string `xorm:"TEXT"`
	ApplyIp   string `xorm:"VARCHAR(50)" json:"ip"`
}

type redisConsumer struct {
	redisClient *redis.Client
	engine      *xorm.Engine
	batchSize   int
	logBatch    []*models.Logs
	mu          sync.Mutex
}

var instance *redisConsumer
var once sync.Once

func GetRedisConsumer() *redisConsumer {
	once.Do(func() {
		redisClient := redis.NewClient(&redis.Options{
			Addr: "localhost:6379",
		})
		engine, _ := xorm.NewEngine("mysql", "root:123456789mk@tcp(localhost:3306)/yami_shops?charset=utf8mb4")
		instance = &redisConsumer{
			redisClient: redisClient,
			engine:      engine,
			batchSize:   100,
			logBatch:    make([]*models.Logs, 0, 100),
		}
		// 只在消费者组不存在时创建
		err := instance.redisClient.XGroupCreate(context.Background(), "adminLogsStream", "your_group", "0").Err()
		if err != nil && err != redis.Nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
			log.Println("Error creating consumer group:", err)
		}
	})
	return instance
}

func (rc *redisConsumer) ConsumeMessagesFromRedisAndStoreInMySQL() {
	for {
		rc.mu.Lock()
		result, err := rc.redisClient.XReadGroup(context.Background(), &redis.XReadGroupArgs{
			Group:    "your_group",
			Consumer: "your_consumer",
			Streams:  []string{"adminLogsStream", ">"},
		}).Result()
		if err != nil {
			log.Println("Error reading from Redis Stream:", err)
			rc.mu.Unlock()
			continue
		}
		for _, stream := range result {
			for _, message := range stream.Messages {
				for _, value := range message.Values {
					logDataJSON, ok := value.(string)
					if !ok {
						continue
					}

					var temp map[string]interface{}
					err := json.Unmarshal([]byte(logDataJSON), &temp)
					if err != nil {
						continue
					}

					tsFloat, ok := temp["ts"].(float64)
					if !ok {
						continue
					}

					var logObj models.Logs
					logObj.Timestamp = int64(tsFloat)
					logObj.LogLevel = temp["level"].(string)
					logObj.Caller = temp["caller"].(string)
					logObj.Msg = temp["msg"].(string)
					logObj.Method = temp["method"].(string)
					logObj.Path = temp["path"].(string)
					logObj.Query = temp["query"].(string)
					if bodyValue, ok := temp["body"]; ok && bodyValue != nil {
						logObj.Body = bodyValue.(string)
					} else {
						logObj.Body = "" // 或者其他合适的默认值
					}
					logObj.ApplyIp = temp["ip"].(string)
					logObj.Username = temp["username"].(string)
					rc.logBatch = append(rc.logBatch, &logObj)

					if len(rc.logBatch) == rc.batchSize {
						// 达到批量大小，进行批量插入
						_, err := rc.engine.Insert(rc.logBatch)
						if err != nil {
							log.Println("Error inserting batch into MySQL:", err)
						}
						rc.logBatch = rc.logBatch[:0]
					}

					// 删除已处理的消息
					err = rc.redisClient.XDel(context.Background(), stream.Stream, message.ID).Err()
					if err != nil {
						log.Println("Error deleting message from Redis Stream:", err)
					}
				}
			}
		}

		// 如果最后一批不足批量大小，也进行插入
		if len(rc.logBatch) > 0 {
			_, err := rc.engine.Insert(rc.logBatch)
			if err != nil {
				log.Println("Error inserting remaining batch into MySQL:", err)
			}
			rc.logBatch = rc.logBatch[:0]
		}
		rc.mu.Unlock()
	}
}
